package com.atguigu.flink.tableapi;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**

 */
public class Demo3_Agg
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());


        Table table = tableEnv.fromDataStream(ds);

        table.printSchema();

        Table result = table
                            .groupBy($("id"))
                            .select($("id"), $("vc").sum().as("sumVC"))
                             ;

        /*
               tableEnv.toDataStream(): 返回的数据流，只能是连续append数据。
                                        如果数据有聚合，意味着数据需要进行更新。
                                        只能消费，只允许insert的表。

                                          s1,vc=1 =====> s1,sumVC=1
                                          s1,vc=3 =====> s1,sumVC=4

                 doesn't support consuming update changes which is produced by node GroupAggregate

              tableEnv.toChangelogStream(result):  返回的数据流可以使用 -U,+U显示表中数据的变化。
                                                    消费一个会发生更新的表。

              tableEnv.toRetractStream(result, Row.class): 消费一个会发生更新的表。
                                提供以boolean标记，标记当前的数据是否是更新后(true)的数据。
         */
        //DataStream<Row> ds2 = tableEnv.toChangelogStream(result);

        DataStream<Tuple2<Boolean, Row>> ds2 = tableEnv.toRetractStream(result, Row.class);

        ds2.filter(t ->t.f0)
           .map(t -> t.f1)
           .print();





        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }




    }
}
